package com.behavioranalysis.flinkprogram.flink.ad;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午9:02
 * updated by yang on 20-2-14 下午9:02
 */
public class DailyCityAdClickCountProcess extends
        ProcessWindowFunction<
                Tuple2<String, Long>,
                Tuple5<String, String, String, Long, Long>,
                String,
                TimeWindow> {
    private static final long serialVersionUID = 1L;
    // 状态
    // TODO 要定期清理 状态
    private transient MapState<String, Long> dailyAdClickCount;

    @Override
    public void open(Configuration parameters) {
        MapStateDescriptor<String, Long> descriptor =
                new MapStateDescriptor<>(
                        "dailyAdClickCount",
                        String.class, Long.class);
        dailyAdClickCount = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void process(String key,
                        Context context,
                        Iterable<Tuple2<String, Long>> elements,
                        Collector<Tuple5<String, String, String, Long, Long>> out)
            throws Exception {
        // 1. 先将key value更新到状态中
        //累加count
        long count = 0L;
        for (Tuple2<String, Long> tuple2 : elements) {
            count += tuple2.f1;
        }
        // 判断此key是否已经存在于状态dailyAdClickCount中
        if (dailyAdClickCount.contains(key)) {
            // 存在，则累加count
            dailyAdClickCount.put(key, dailyAdClickCount.get(key) + count);
        } else {
            // 不存在，则put
            dailyAdClickCount.put(key, count);
        }
        // 2. 将此key在状态中的value得到并且返回
        String[] keySplited = key.split("_");
        String date = keySplited[0];
        String province = keySplited[1];
        String city = keySplited[2];
        long adid = Long.valueOf(keySplited[3]);
        out.collect(new Tuple5<>(date, province, city, adid, dailyAdClickCount.get(key)));

        /**
         * 每天每个省份每个城市对某个广告的点击量
         * 集群中的内存中保存了一份（状态）：集群中的我们要找到实时的每个省份的top3广告
         * mysql数据库中也保存了一份：数据库中的我们要在前端展示
         *
         * 将全局的状态设置为 侧面输出流
         *
         * 这个方法是 一个窗口中的每个key 都要执行一次
         * 那说明，每个key里面都要将全局的状态得到
         * 然后再去计算每个省份的top3，有点频繁。
         * TODO 能不能每个窗口计算一次，而不是每个key
         *
         * 并且计算实时的每个省份的top3广告，然后保存到数据库
         */

//                        for (Map.Entry<String, Long> entry : dailyAdClickCount.entries()) {
//                            context.output(dailyAdClickCountGlobal, new Tuple2<>(entry.getKey(), entry.getValue()));
//                        }
    }
}
